package bolts;

import java.io.IOException;
import java.util.Map;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import com.fasterxml.jackson.databind.ObjectMapper;

public class FilterNonEnglishBolt extends BaseBasicBolt {

    private final ObjectMapper jsonMapper = new ObjectMapper();

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("screenName", "text", "createdAt"));
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        try {
            final Map<String, Object> data = jsonMapper.readValue(input.getString(0), Map.class);
            if (data.containsKey("created_at") && "en".equalsIgnoreCase((String) data.get("lang"))) {
                collector.emit(new Values(((Map<String, Object>) data.get("user")).get("screen_name"), data.get("text"), data.get("created_at")));
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
